package com.example.message;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

//实现响应式编程,适合用于构建高性能、高并发的应用。Flux代表一个包含0到N个元素的异步序列,异步的、非阻塞的数据流.与Java中的Stream不同，Flux的元素可以按需生成，而不需要一次性加载到内存中。这使得Flux非常适合处理大量数据或需要实时响应的场景
//Flux提供了一系列的操作符（operators），用于转换、过滤、组合和聚合数据流。这些操作符是响应式编程的核心，它们允许你以声明式的方式描述数据流的处理逻辑。
//Flux 可以触发零到多个事件，并根据实际情况结束处理或触发错误。
//Mono 最多只触发一个事件，所以可以把 Mono 用于在异步任务完成时发出通知。

// Flux 和 Mono 都是数据流的发布者，使用 Flux 和 Mono 都可以发出三种数据信号：元素值，错误信号，完成信号；错误信号和完成信号都代表终止信号，终止信号用于告诉订阅者数据流结束了，错误信号终止数据流同时把错误信息传递给订阅者。
//三种信号的特点：
//错误信号和完成信号都是终止信号，不能共存
//如果没有发送任何元素值，而是直接发送错误或者完成信号，表示是空数据流
//如果没有错误信号，也没有完成信号，表示是无限数据流
public class FluxExample {
    public static void main(String[] args) {
        mono();
    }

    public static void create() {
        //just()：创建Flux序列，并声明数据流，
        Flux<Integer> integerFlux = Flux.just(1, 2, 3, 4);//整形
        //subscribe()：订阅Flux序列，只有进行订阅后才回触发数据流，不订阅就什么都不会发生
        integerFlux.subscribe(System.out::println);

        Flux<String> stringFlux = Flux.just("hello", "world");//字符串
        stringFlux.subscribe(System.out::println);

        //fromArray(),fromIterable()和fromStream()：可以从一个数组、Iterable 对象或Stream 对象中创建Flux序列
        Integer[] array = {1,2,3,4};
        Flux.fromArray(array).subscribe(System.out::println);

        List<Integer> integers = Arrays.asList(array);
        Flux.fromIterable(integers).subscribe(System.out::println);

        Stream<Integer> stream = integers.stream();
        Flux.fromStream(stream).subscribe(System.out::println);
    }

    public static void flux() {
        // 方法一 创建一个Flux，包含一些整数元素
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);

        // 转换与映射 使用map操作符将每个元素乘以2
        Flux<Integer> doubledFlux = flux.map(n -> n * 2);

        // 订阅Flux并处理每个元素
        doubledFlux.subscribe(System.out::println);


        // 方法二
        List<String> list = Arrays.asList("D", "E", "F");
        Flux<String> fromList = Flux.fromIterable(list);

        // 方法三
        //Flux.generate(Supplier<SynchronousSink<T>>)、Flux.fromStream(Stream<T>) 或 Flux.from(Publisher<T>)。

        Flux<String> mappedFlux = fromList.map(String::toUpperCase);

        // 过滤
        Flux<String> filteredFlux = mappedFlux.filter(s -> s.startsWith("A"));

        // 组合
        Flux<String> concatFlux = filteredFlux.concatWith(Flux.just("G", "H"));

        // 聚合操作
        //Flux<String> reducedFlux = flux.reduce((s1, s2) -> s1 + ", " + s2);

        concatFlux.subscribe(
                value -> System.out.println("Received: " + value),
                error -> System.out.println("Error: " + error.getMessage()),
                () -> System.out.println("Completed")
        );


        //窗口与缓冲
        //使用 window(int)、buffer(int) 分割 Flux 为多个子序列。

        //背压支持
        //Reactor 自动处理背压，你可以通过设置 limitRate()、take(int) 等方法限制速率或数量。

        //终端操作
        //使用 blockFirst()、blockLast()、toStream() 或 collectList() 等方法等待 Flux 结果并获取它。

        //定时与延迟
        //使用 delayElements(Duration) 或 interval(Duration) 为发出元素设定延迟。

        //条件与分支
        //使用 switchIfEmpty()、defaultIfEmpty() 等方法根据 Flux 是否为空进行不同操作。

        Flux<String> numbers = Flux.range(1, 5)
                .map(Object::toString)
                .filter(s -> Integer.parseInt(s) % 2 == 0)
                .doOnNext(System.out::println)
                .delayElements(Duration.ofMillis(100));
        numbers.subscribe();

        //错误处理
        //使用 onErrorReturn(T)、onErrorResume(Function) 或 doOnError(Consumer) 处理错误情况。
    }

    public static void m() {
        Flux<Integer> flux1 = Flux.just(1, 2, 3);
        Flux<Integer> flux2 = Flux.just(4, 5, 6);
        // 将两个Flux按顺序合并，合并后的Flux中先包含第一个Flux的所有元素，然后再包含第二个Flux的所有元素。如果两个Flux中存在相同的元素，合并后的Flux中会保留其中一个
        Flux<Integer> mergedFlux1 = flux1.concatWith(flux2);

        // 将两个Flux按并行方式合并，合并后的Flux中包含两个Flux的所有元素，如果两个Flux中存在相同的元素，合并后的Flux中会保留所有相同元素
        Flux<Integer> mergedFlux2 = flux1.mergeWith(flux2);

        // 将两个Flux按索引位置一一对应合并，合并后的Flux中包含两个Flux中对应位置的元素。如果两个Flux的长度不一致，合并后的Flux的长度将取决于较短的那个Flux
        Flux<Tuple2<Integer, Integer>> mergedFlux = flux1.zipWith(flux2);
    }

    // Mono常用于表示单个值的异步计算或异步操作的结果。
    // 在实际应用中，你可能会用Mono来封装数据库查询、网络请求或其他可能返回单个结果的异步操作。Mono提供了丰富的操作符，用于转换、过滤和组合单个值的异步流。
    public static void mono() {
        // 创建一个Mono，它将在1秒后发出一个值"Hello, Mono!"
        Mono<String> mono = Mono.just("Hello, Mono!")
                .delayElement(Duration.ofSeconds(1));

        // 订阅Mono并处理它的值
        mono.subscribe(System.out::println);

        // 由于Mono是异步的，主线程不会等待Mono完成。
        // 为了防止程序立即退出，我们在这里添加一个延迟。
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 容错
        Mono.just("java")
                .concatWith(Mono.error(new IllegalStateException()))
                .onErrorReturn("default")
          .subscribe(System.out::println);

        Flux.just("javaedge1", "javaedge2", "javaedge3").subscribe(data -> System.out.println("onNext:" + data), err -> {
        }, () -> System.out.println("onComplete"));
    }
}
